package cn.monkey.io.transport.netty.client;

import cn.monkey.io.transport.core.Exchange;
import cn.monkey.io.transport.core.client.Client;
import cn.monkey.io.transport.netty.*;
import cn.monkey.io.transport.netty.reactive.ReactiveNettyExchangeManager;
import cn.monkey.io.transport.netty.utils.ChannelUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.SocketChannel;
import io.netty.util.AttributeKey;
import io.netty.util.concurrent.Future;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;

public class ClientTransport extends Transport<Bootstrap, ClientTransportConfig, ClientTransportConfig.Builder, Channel> implements Client<NettyExchange> {

    private static final Logger log = LoggerFactory.getLogger(ClientTransport.class);

    protected final NettyExchangeManager<NettyExchange> nettyExchangeManager;

    protected final ReactiveNettyExchangeManager<NettyExchange> reactiveNettyExchangeManager;

    public ClientTransport(ClientTransportConfig config, NettyExchangeManager<NettyExchange> nettyExchangeManager) {
        super(config);
        this.nettyExchangeManager = nettyExchangeManager;
        this.reactiveNettyExchangeManager = null;
    }

    public ClientTransport(ClientTransportConfig config, ReactiveNettyExchangeManager<NettyExchange> reactiveNettyExchangeManager) {
        super(config);
        this.reactiveNettyExchangeManager = reactiveNettyExchangeManager;
        this.nettyExchangeManager = null;
    }

    @Override
    protected ClientTransportConfig complementConfig(ClientTransportConfig config) {
        ShutDown transportShutDown = config.shutDown();
        if (transportShutDown == null) {
            return config.mutate().shutDown(transport -> {
                List<Mono<Void>> monoList = this.channelMap.values()
                        .stream()
                        .map(ChannelOutboundInvoker::close)
                        .map(channelFuture -> Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
                            if (future.isSuccess()) {
                                sink.success();
                                return;
                            }
                            sink.error(future.cause());
                        })).then()).toList();
                ArrayList<Mono<Void>> monos = new ArrayList<>(monoList);
                Future<?> objectFuture = shutDownEventLoopGroup(config.eventLoopGroup());
                monos.add(Mono.create(sink -> objectFuture.addListener(future -> {
                    if (future.isSuccess()) {
                        sink.success();
                        return;
                    }
                    sink.error(future.cause());
                })));
                return Flux.merge(monos).then();
            }).build();
        }
        return config;
    }

    @Override
    protected Bootstrap newBootStrap() {
        return new Bootstrap();
    }

    @Override
    protected Bootstrap doOnChannelInit(Bootstrap bootstrap) {
        Consumer<Channel> channelCustomizer = this.config.channelInitCustomizer();
        return bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                if (channelCustomizer != null) {
                    channelCustomizer.accept(ch);
                }
            }
        });
    }

    protected String key(Channel channel) {
        return ChannelUtil.getId(channel);
    }

    @Override
    public Mono<NettyExchange> connect(SocketAddress remoteAddress) {
        if (this.nettyExchangeManager != null) {
            return this.connect0(remoteAddress);
        }
        if (this.reactiveNettyExchangeManager != null) {
            return this.reactiveConnect(remoteAddress);
        }
        return Mono.error(new IllegalStateException("no active exchange manager"));
    }

    private Mono<NettyExchange> reactiveConnect(SocketAddress remoteAddress) {
        ChannelFuture channelFuture = this.bootstrap.connect(remoteAddress);
        Channel channel = this.channelMap.compute(remoteAddress, (socketAddress, c) -> {
            if (c == null || !c.isActive()) {
                return channelFuture.channel();
            }
            return c;
        });
        return Mono.create(sink -> channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                sink.success();
                return;
            }
            sink.error(future.cause());
        })).flatMap(v -> this.reactiveNettyExchangeManager.findOrCreate(key(channel), channel));
    }

    private Mono<NettyExchange> connect0(SocketAddress remoteAddress) {
        ChannelFuture channelFuture = this.bootstrap.connect(remoteAddress);
        Channel channel = this.channelMap.compute(remoteAddress, (address, c) -> {
            if (c == null || !c.isActive()) {
                return channelFuture.channel();
            }
            return c;
        });
        return Mono.create(sink -> channelFuture.addListener((ChannelFutureListener) future -> {
            if (future.isSuccess()) {
                sink.success(this.nettyExchangeManager.findOrCreate(key(channel), channel));
                return;
            }
            sink.error(future.cause());
        }));
    }

    @Override
    public Mono<Void> close(SocketAddress socketAddress) {
        Channel channel = this.channelMap.remove(socketAddress);
        if (channel == null) {
            log.warn("socket channel is not exists");
            return Mono.error(new NullPointerException());
        }
        AttributeKey<NettyExchange> attributeKey = AttributeKey.valueOf(Exchange.class, Exchange.KEY);
        NettyExchange nettyExchange = channel.attr(attributeKey).get();
        if (nettyExchange == null) {
            return Mono.error(new IllegalStateException("netty exchange not bind yet"));
        }
        if (this.nettyExchangeManager != null) {
            try {
                this.nettyExchangeManager.destroy(nettyExchange);
                return Mono.empty();
            } catch (Throwable e) {
                return Mono.error(e);
            }
        }
        if (this.reactiveNettyExchangeManager != null) {
            return this.reactiveNettyExchangeManager.destroy(nettyExchange);
        }
        return Mono.error(new IllegalStateException("no active exchange manager"));
    }

    @Override
    public NettyExchange getExchange(SocketAddress remoteAddress) {
        Channel channel = this.channelMap.get(remoteAddress);
        if (channel == null) {
            throw new NullPointerException("can not find exchange before execute #connect(SocketAddress)");
        }
        AttributeKey<NettyExchange> attributeKey = AttributeKey.valueOf(Exchange.class, Exchange.KEY);
        NettyExchange nettyExchange = channel.attr(attributeKey).get();
        if (nettyExchange != null) {
            return nettyExchange;
        }
        throw new NullPointerException("netty exchange has not bind yet!");
    }
}
